package com.openatc.agent.gateway;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.openatc.agent.config.WebSocketConfig;
import com.openatc.agent.model.WebSocketSub;
import com.openatc.agent.utils.ApplicationContextProvider;
import com.openatc.agent.utils.TokenUtil;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

import static com.openatc.core.common.IErrorEnumImplOuter.E_3015;

/**
 * @author laoxia
 * @Date 2021/7/28 10:20
 * @Describetion
 */
@ServerEndpoint(value = "/ws")
@Component
@Data
public class WebSocketServer {

    private Logger log = LoggerFactory.getLogger(WebSocketServer.class);
    private String EventAgentFault = "status/fault"; //设备故障消息类型
    private String StatusPattern = "status/pattern"; //方案相位消息类型
    private String EventTrafficData = "event/trafficdata"; //交通事件
    private String TokenVerify = "token/verify"; // token验证消息

    //当前在线总数
    private static Integer onlineCount = 0;

    private final static Map<Session, WebSocketServer> patternWebSocketSet = new ConcurrentHashMap<>();
    private final static Map<Session, WebSocketServer> trafficIncidentWebSocketSet = new ConcurrentHashMap<>();
    private final static Map<Session, WebSocketServer> faultIncidentWebSocketSet = new ConcurrentHashMap<>();
    public Set<String> patternAgentidSet = new CopyOnWriteArraySet<>(); // 当前客户端订阅方案的路口ID
    public Set<String> authAgentidSet = new HashSet<>(); // 当前客户端用户拥有权限的路口

    // 此处存放Spring boot创建的redisService
    public static WebSocketServer webSocketComponent;

    //与客户端的连接会话
    private Session session;

    @Autowired
    private RedisService redisService;

    Gson gson = new Gson();

    // session中保存当前session是否通过token验证的key
    private static final String TOKEN_PASS_VERIFY_KEY = "tokenVerify";

    @PostConstruct
    public void init() {
        WebSocketServer.webSocketComponent = this;
    }

    @OnOpen
    public void onOpen(Session session) {
        synchronized (onlineCount) {
            this.session = session;
            this.session.setMaxIdleTimeout( 60 * 1000); // 一分钟无消息自动关闭ws
            this.session.setMaxTextMessageBufferSize( 1024 * 1024);
            session.getUserProperties().put(TOKEN_PASS_VERIFY_KEY,false);

            onlineCount++;
            log.info("onOpen URL:{},Session ID:{},OnlineCount:{}", session.getRequestURI(), session.getId(), onlineCount);
        }
    }

    @OnMessage
    public void onMessage(String message) {
        log.info("onMessage sessionId:{},receive:{}", session.getId(), message);

        WebSocketSub wss = gson.fromJson(message, WebSocketSub.class);
        if(wss == null){
            log.info("onMessage error fromJson msg: " + message);
            return;
        }


        int processFlag = processToken(wss);
        if (processFlag == -1) {
            return;
        }

        String infoType = wss.getInfotype().trim(); //订阅类型
        String subscribe = wss.getSubscribe(); //开始or结束标
        String[] para = wss.getParam(); //订阅的通道
        String model = wss.getModel().trim();

        if (infoType.isEmpty() || subscribe.isEmpty()) {
            log.info("onMessage error: format empty");
            return;
        }

        if ("up".equals(subscribe)) {//开启订阅消息
            // 拥堵事件订阅消息
            if (EventTrafficData.equals(infoType)) {
//                if (trafficIncidentWebSocketSet.size() == 0)
                    webSocketComponent.redisService.subsMessage(model + ":" + EventTrafficData + ":*");
                trafficIncidentWebSocketSet.put(session , this);
            }
            // 故障事件订阅消息
            else if (EventAgentFault.equals(infoType)) {
//                if (faultIncidentWebSocketSet.size() == 0)
                    webSocketComponent.redisService.subsMessage(model + ":" + EventAgentFault + ":*");
                faultIncidentWebSocketSet.put(session ,  this);
            // 方案状态
            } else if (StatusPattern.equals(infoType)) {
//                if (patternWebSocketSet.size() == 0)
//                    webSocketComponent.redisService.subsMessage(model + ":" + StatusPattern + ":*");
                patternWebSocketSet.put(session ,  this);

                //添加监听路口映射,这里 null 代表为管理员，不做限制
                for (int i = 0; i < para.length; i++) {
                    // 验证token，根据用户权限添加发送消息路口
                    if ( WebSocketConfig.TOKEN_VERIFY_FLAG) {
                        // 此处null代表管理员角色，无限制
                        if (authAgentidSet== null || authAgentidSet.contains(para[i]))
                            patternAgentidSet.add(para[i]);
                    }
                    else
                        patternAgentidSet.add(para[i]);
                }
                // 订阅后立即应答当前方案状态
//                try {
//                    RedisService redisService = webSocketComponent.redisService;
//                    redisService.sendMessageToSubsClients(infoType);
//                } catch (Exception e) {
//                    log.info("RedisConnectionException: Unable to connect to redis:6379");
//                    return;
//                }
            }

        } else if ("down".equals(subscribe)) {//结束消息订阅
            // 事件消息
            if (EventTrafficData.equals(infoType)) {
                trafficIncidentWebSocketSet.remove(session);
                if (trafficIncidentWebSocketSet.size() == 0)
                    webSocketComponent.redisService.unSubsMessage(model + ":" + EventTrafficData + ":*");
            }
            //事故消息
            if (EventAgentFault.equals(infoType)) {
                faultIncidentWebSocketSet.remove(session);
                if (faultIncidentWebSocketSet.size() == 0)
                    webSocketComponent.redisService.unSubsMessage(model + ":" + EventAgentFault + ":*");
            }
            //方案消息
            if (StatusPattern.equals(infoType)) {
                for (int i = 0; i < para.length; i++) {
                    patternAgentidSet.remove(para[i]);
                }
                if(patternAgentidSet.size() == 0)
                    patternWebSocketSet.remove(session);
//                    webSocketComponent.redisService.unSubsMessage(model + ":" + StatusPattern + ":*");
            }

        } else {
            log.info("Unkonw Subscribe Type!");
        }
    }

    /**
     * 处理token
     * @param wss
     * @return
     */
    private int processToken(WebSocketSub wss) {
        String infoType = wss.getInfotype();
        String token = wss.getParam()[0];
//        String model = wss.getModel();

        // 配置了不验证ws的token
        if (!WebSocketConfig.TOKEN_VERIFY_FLAG) {
            // 消息类型为token/verify，不处理
            if (TokenVerify.equals(infoType)) {
                return -1;
            }
            // 消息类型为其他,放行
            else {
                return 0;
            }
        }
        String resMsg = "";
        if (!StringUtils.hasLength(token)) {
            resMsg = getResMsg(false, infoType);
            sendMessage(resMsg);
            return -1;
        }

        // 如果需要结束onMessage方法，返回-1
        TokenUtil tokenUtil = ApplicationContextProvider.getBean(TokenUtil.class);
        // 1、token验证消息
        if (TokenVerify.equals(infoType)) {
//            String gentoken = TokenController.protokenMap.get(token);
//            if (StringUtils.hasLength(gentoken)) {
//                token = gentoken;
//            }

            boolean tokenPass = false;
            try{
                tokenPass = tokenUtil.validateToken(token);
            }
            catch (Exception e){
                log.warn("WebSocket token 校验失败: " + e.getMessage());
            }
            if(tokenPass){
                authAgentidSet = tokenUtil.getDevsFromToken(token);
            }
            // 成功:应答验证成功消息,改变session标志位
            session.getUserProperties().put(TOKEN_PASS_VERIFY_KEY,tokenPass);
            resMsg = getResMsg(tokenPass, infoType);
            sendMessage(resMsg);
            return -1;
        }
        //2、其他订阅消息
        else {
            boolean tokenPass = (boolean) session.getUserProperties().get(TOKEN_PASS_VERIFY_KEY);
            // token标志位为true：不做处理
            if (!tokenPass) {
                // token标志位为true：应答3015
                sendMessage(getResMsg(false, infoType));
                return -1;
            }
        }
        return 0;
    }

    private String getResMsg(Boolean validateToken, String infoType) {
        JsonObject resMsg = new JsonObject();
//        resMsg.addProperty("model",model);
        resMsg.addProperty("infotype",infoType);
        resMsg.addProperty("success",validateToken);
        if (validateToken) {
            resMsg.addProperty("code","0");
            resMsg.addProperty("message","success");
        } else {
            resMsg.addProperty("code",E_3015.getErrorCode());
            resMsg.addProperty("message",E_3015.getErrorMsg());
        }
        return resMsg.toString();
    }

    @OnError
    public void onError(Throwable error) throws IOException {
        log.error("onError:{}，Session ID:{}", error.toString(), session.getId());
        session.close();
    }


    @OnClose
    public void onClose(Session session, CloseReason reason) {
        patternWebSocketSet.remove(session);
//        if (patternWebSocketSet.size() == 0)
//            webSocketComponent.redisService.unSubsMessage("asc" + ":" + StatusPattern + ":*");

        trafficIncidentWebSocketSet.remove(session);
        if (trafficIncidentWebSocketSet.size() == 0)
            webSocketComponent.redisService.unSubsMessage("asc" + ":" + EventTrafficData + ":*");

        faultIncidentWebSocketSet.remove(session);
        if (faultIncidentWebSocketSet.size() == 0)
            webSocketComponent.redisService.unSubsMessage("asc" + ":" + EventAgentFault + ":*");

        synchronized (onlineCount) {
            onlineCount--;
            log.info("onClose:ID: {}, {}, OnlineCount：{}", session.getId(), reason.toString(), onlineCount);
        }
//        }
    }


    /**
     * 整理前端发送的数据集合
     *
     * @return
     */
//    public void setInfoAndAgentIds(String[] para, String infoType, String subscribe) {
//        Set<String> agentIds = this.infoAndAgentIds.get(infoType.trim());
//        if (agentIds == null) agentIds = new HashSet<>();
//        for (String value : para) {
//            if (value == null) continue;
//            String agentId = value.trim();
//            if (subscribe.equals("up")) {
//                agentIds.add(agentId);
//            } else if (subscribe.equals("down")) {
//                agentIds.remove(agentId);
//            }
//        }
//        this.infoAndAgentIds.put(infoType.trim(), agentIds);
//    }


    /**
     * 发送消息
     *
     * @param message
     */
    public synchronized void sendMessage(String message) {
        try {
            session.getBasicRemote().sendText(message);
        } catch (Exception e) {
            log.error("sendMessage Error:{},session ID:{}", e.getMessage(), session.getId());
        }
    }


    public static Map<Session, WebSocketServer> getPatternWebSocketSet() {
        return patternWebSocketSet;
    }

    public static Map<Session, WebSocketServer> getTrafficIncidentWebSocketSet() {
        return trafficIncidentWebSocketSet;
    }

    public static Map<Session, WebSocketServer> getFaultIncidentWebSocketSet() {
        return faultIncidentWebSocketSet;
    }

}
